package com.simple.datax.conn.impl;

import java.util.Hashtable;

import javax.jms.MessageListener;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.ibm.mq.MQC;
import com.ibm.mq.MQException;
import com.ibm.mq.MQGetMessageOptions;
import com.ibm.mq.MQMessage;
import com.ibm.mq.MQPutMessageOptions;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;
import com.simple.datax.SimpleMessage;
import com.simple.datax.api.ConnectorException;

/** 
 * @Title: IBMMQConnImpl.java 
 * @Package com.simple.datax.conn.impl 
 * @Description: 实现具体的IBM MQ连接操作，包括连接，接收，发送等。 
 * @author Jin Yi 
 * @date 2012-12-30
 * @version V1.0 
 */
public class IBMMQConnImpl extends MQBaseConnector {
	private static final Logger logger = LoggerFactory
			.getLogger(IBMMQConnImpl.class);
	public static final String TYPE = "IBMMQ";

	private String host;
	private String port;
	private String ccsid;
	private String channel;
	private MQQueue queue;
	private String manager;
	private String queuename;
	private String username;
	private String password;
	private MQQueueManager qMgr;
	private MQGetMessageOptions mgo;
	private MQPutMessageOptions mqPutMessageOptions;
	private Hashtable<String,Object> prop;
	
	private long recvTotal;
	private long sendTotal;
	
	/**
	 * 构造器
	 * @param connPara
	 * @throws ConnectorException
	 */
	public IBMMQConnImpl(String connPara) throws ConnectorException {
		super(connPara);
		this.connPara = connPara;
		try {
			parseConnPara();
			createMQ();
		} catch (Exception e) {
			throw new ConnectorException("Create IBMMQConnImpl failed", e);
		}
	}

	/**
	 * 创建IBM MQ对象
	 * @throws Exception
	 */
	private void createMQ() throws Exception {
		prop = new Hashtable<String, Object>();
		prop.put(MQC.PORT_PROPERTY, new Integer(port));
		prop.put(MQC.HOST_NAME_PROPERTY, host);
		prop.put(MQC.CHANNEL_PROPERTY, channel);
//		prop.put(MQC.TRANSPORT_PROPERTY, MQC.TRANSPORT_MQSERIES);
		prop.put(MQC.CCSID_PROPERTY, new Integer(ccsid));
		
		if (!username.trim().isEmpty()) {
			prop.put(MQC.USER_ID_PROPERTY, username);
			prop.put(MQC.PASSWORD_PROPERTY, password.trim());
		}

//		MQEnvironment.hostname = host; // 服务器ip地址
//		MQEnvironment.port = Integer.parseInt(port);
//		MQEnvironment.CCSID = Integer.parseInt(ccsid); // // 1381表示是简体中文，
//		// CCSID的值在AIX上一般设为1383，如果要支持GBK则设为1386，在WIN上设为1381。
//		// 服务器MQ服务使用的编码1381代表GBK、1208代表 UTF(Coded Character Set
//		// Identifier:CCSID)
//		MQEnvironment.channel = channel; // 服务器连接通道名
////		MQEnvironment.userID = username;
////		MQEnvironment.password = password;
	}
	
	/**
	 * 为接收消息做准备
	 * @throws ConnectorException
	 */
	private void prepareReceiveQueue() throws ConnectorException {
		if (qMgr == null || queue ==  null || !qMgr.isOpen()
				|| !qMgr.isConnected() || !queue.isOpen()) {
			disconnect();
			try {
				qMgr = new MQQueueManager(manager, prop);
				int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_FAIL_IF_QUIESCING
						| MQC.MQOO_INQUIRE;
				// int openOptions = MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING |
				// MQC.MQOO_INQUIRE ;
				queue = qMgr.accessQueue(queuename, openOptions);
				mgo = new MQGetMessageOptions();
				mgo.options |= MQC.MQGMO_NO_WAIT;
			} catch (MQException e) {
				throw new ConnectorException("IBMMQ initialize failed", e);
			}
		}
	}

	@Override
	public void setMessageListener(MessageListener listener)
			throws ConnectorException {
		// 目前所有队列采用轮询机制，所以没有实现监听机制。
	}

	@Override
	public String getMessage() throws ConnectorException {
		String ret = null;
		prepareReceiveQueue();
		MQMessage msg = new MQMessage();
		try {
			// 取消息
			queue.get(msg, mgo);
			if (msg != null) {
				byte[] xmlData = new byte[msg.getDataLength()];
				msg.readFully(xmlData);
				ret = new String(xmlData, getEncoding());
			}
			logger.debug("########Receiver Get Message: #" + ++recvTotal + " - " + connPara);
		} catch (MQException ex) {
			// Reason Code - 2033  表示队列中没有数据
			if (ex.reasonCode != 2033) {
				logger.error("########Receiver Get Message Failed: #" + ++recvTotal + " - " + connPara);
				throw new ConnectorException("Read messge from MQ failed.", ex);
			}
		} catch (Exception e) {
			logger.error("########Receiver Get Messag Failed: #" + ++recvTotal + " - " + connPara);
			throw new ConnectorException("Read messge from MQ failed.", e);
		}
		return ret;
	}
	
	/**
	 * 解析连接参数
	 * @throws ConnectorException 
	 */
	private void parseConnPara() throws ConnectorException {
		//connPara = "HOST$PORT$USER$PASSWORD$CCSID$MANAGER$CHANNEL$QUEUE";
//		"10.2.3.61$1500$mqadmin$Art515940$1381$QMRTSC$JAVA.CHANNEL$QTSC.EC0001.RECE";
//		"10.2.3.61$1502$mqadmin$Art515940$1381$QMLTSC$JAVA.CHANNEL$QTSC.RECE.EC9999999999";
		String[] tmpStr = splitConnPara(connPara);
		if (tmpStr.length==8) {
			host = tmpStr[0];
			port = tmpStr[1];
			if (!port.matches("[\\d]+")) {
				throw new ConnectorException("IBMMQ Connector Port is not number -" + port);
			}
			username = tmpStr[2];
			password = tmpStr[3];
			ccsid = tmpStr[4];
			if (!ccsid.matches("[\\d]+")) {
				throw new ConnectorException("IBMMQ Connector CCSID is not number - " + ccsid);
			}
			manager = tmpStr[5];
			channel = tmpStr[6];
			queuename = tmpStr[7];
		} else {
			throw new ConnectorException(
					"Parsing IBMMQ Connector parameters failed. Please check parameter - "
							+ connPara);
		}
	}
	
	/**
	 * 为发送消息做准备
	 * @throws ConnectorException
	 */
	private void prepareSendQueue() throws ConnectorException {
		if (qMgr == null || queue ==  null || !qMgr.isOpen()
				|| !qMgr.isConnected() || !queue.isOpen()) {
			disconnect();
			try {
				int sendOptions = MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING;
				qMgr = new MQQueueManager(manager, prop);
				queue = qMgr.accessQueue(queuename, sendOptions, null, null, null);
				mqPutMessageOptions = new MQPutMessageOptions();
			} catch (MQException e) {
				throw new ConnectorException("IBMMQ initialize failed", e);
			}
		}
	}
	
	@Override
	public String getConnType() {
		return TYPE;
	}
	
	@Override
	protected void prepareSend() throws ConnectorException {
		prepareSendQueue();
	}

	@Override
	protected void internalSend(SimpleMessage message) throws ConnectorException {
		MQMessage mqMessage = null;
		try {
			mqMessage = new MQMessage();
			// TODO setup a mapping of encoding name to encoding code 
//			mqMessage.encoding = 546;
			mqMessage.characterSet = Integer.parseInt(ccsid);
			mqMessage.format = MQC.MQFMT_STRING;
//			String content = getOutputBytes(message.getContent(), getEncoding());
			mqMessage.write(getOutputBytes(message.getContent(), getEncoding()));
//			mqMessage.writeString(content);
			queue.put(mqMessage, mqPutMessageOptions);
			qMgr.commit();
			logger.debug("########Sender Sent Message: #" + ++sendTotal + " - " + connPara);
		} catch (MQException ex) {
			logger.error("########Sender Sent Message Failed: #" + ++sendTotal + " - " + connPara);
			throw new ConnectorException("IBMMQ internalSend failed", ex, true);
		} catch (Exception ex) {
			logger.error("########Sender Sent Message Failed: #" + ++sendTotal + " - " + connPara);
			throw new ConnectorException("IBMMQ internalSend failed", ex);
		}
	}
	
	public static void main(String[] args) {
//		String connPara = "10.2.3.20$1502$mqadmin$Art515940$1381$QMRECA$CHL.TSC.EC0001$QTSC.SEND.TRANSFER";
		String connPara = "10.2.3.20$1502$ $ $1381$QMLTSC$JAVA.CHANNEL$QECA.RECE.LOCAL";
		try {
			IBMMQConnImpl conn = new IBMMQConnImpl(connPara);
			SimpleMessage object = new SimpleMessage();
			object.setContent("test_test");
//			String message = conn.getMessage();
			conn.send(object);
//			System.out.println(message);
		} catch (ConnectorException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void disconnect() {
		if (queue != null) {
			try {
				queue.close();
			} catch (MQException e) {
				logger.warn("IBMMQ disconnect error", e);
			}
			qMgr = null;
		}
		
		if (qMgr != null) {
			if (qMgr.isConnected()) {
				try {
					qMgr.disconnect();
				} catch (MQException e) {
					logger.warn("IBMMQ disconnect error", e);
				}
			}
			if (qMgr.isOpen()) {
				try {
					qMgr.close();
				} catch (MQException e) {
					logger.warn("IBMMQ disconnect error", e);
				}
			}
			qMgr = null;
		}
		
	}

	@Override
	protected void prepareReceive() throws ConnectorException {
		prepareReceiveQueue();
	}
}
